package cn.boat.router.consumer;


import cn.boat.router.models.Broker;
import cn.boat.router.models.BrokerModel;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CountDownLatch;

// 该方法是得到当前consumer 当前已经订阅过的topic还有相关信息
public class GetConsumerMetaDataImpl {
    private static final String BROKERS = "/boatMQ/brokers";
    private static final String TOPICS = "/topics";
    private static final String PARTITIONS = "/partitions";
    private static final String CONSUMERS = "/boatMQ/consumers";
    private static final String IDS = "/ids";
    private static final String OWNERS = "/owners";
    private static final String STATE = "/state";
    public ConsumerMetaData getConsumerMetaData (String consumerGroupId, String consumerId, ZooKeeper zooKeeperConnection) throws Exception {
        List<TopicWithPartitionData> topicWithPartitionDataList = new ArrayList<>();
        List<String> alreadytopicList = zooKeeperConnection.getChildren(CONSUMERS + "/" + consumerGroupId + OWNERS, false);
        Collections.sort(alreadytopicList);
        ConsumerMetaData consumerMetaData = new ConsumerMetaData();
        if (alreadytopicList.size()!=0){
            for (String alreadyTopic : alreadytopicList) {
                TopicWithPartitionData topicWithPartitionData = new TopicWithPartitionData();
                List<PartitionData> partitionDataList = new ArrayList<>();
                List<String> alreadytopicParList = zooKeeperConnection.getChildren(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + alreadyTopic, false);
                //静态方法sort对列表进行排序
                Collections.sort(alreadytopicParList);
                for (String alreadytopicParNum : alreadytopicParList) {
                    //分区编号，获取为字符形式，存储为整型形式
                    byte[] subParConsumerId = zooKeeperConnection.getData(CONSUMERS + "/" + consumerGroupId + OWNERS + "/" + alreadyTopic + "/" + alreadytopicParNum, false, null);
                    //如果括号里是逻辑表达式不成立，则不执行后面语句
                    //Objects.equals(a, b)比较时, 若a 和 b 都是null, 则返回 true
                    if (!Objects.equals(consumerId,new String(subParConsumerId))){
                        continue;
                    }
                    PartitionData partitionData =new PartitionData();
                    if (Objects.equals(consumerId,new String(subParConsumerId))){
                        partitionData.setPartitionNumber(Integer.valueOf(alreadytopicParNum));
                        byte[] partitionLeader = zooKeeperConnection.getData(BROKERS + TOPICS + "/" + alreadyTopic + PARTITIONS + "/" + alreadytopicParNum + STATE, false, null);
                        //反序列化
                        BrokerPartitionState brokerPartitionState = (BrokerPartitionState) JSONUtils.deserializeObject(new String(partitionLeader), BrokerPartitionState.class);
                        int leader = brokerPartitionState.getLeader();
                        List<Integer> isr =new ArrayList<>();
                        isr = brokerPartitionState.getIsr();
                        byte[] brokerModelData = zooKeeperConnection.getData(BROKERS + IDS + "/" + leader, false, null);
                        BrokerModel brokerModel =(BrokerModel)JSONUtils.deserializeObject(new String(brokerModelData),BrokerModel.class);
                        partitionData.setLeaderIp(brokerModel.getBrokerHost());
                        partitionData.setLeaderPort(brokerModel.getBrokerPort());
                        partitionData.setPartitionLeader(leader);
                        partitionData.setIsr(isr);
                        partitionDataList.add(partitionData);

                    }
                }
                if (partitionDataList.size()!=0){
                    topicWithPartitionData.setTopic(alreadyTopic);
                    topicWithPartitionData.setPartitionDataList(partitionDataList);
                    topicWithPartitionDataList.add(topicWithPartitionData);
                }

            }
        }
        consumerMetaData.setConsumerGroupName(consumerGroupId);
        consumerMetaData.setConsumerId(consumerId);
        consumerMetaData.setTopicWithPartitionDataList(topicWithPartitionDataList);

        return consumerMetaData;
    }
    public List<ConsumerMetaData> getConsumerMetaData() throws Exception {
        String zookeeperId= "";
        GetZookeeperProperties getZookeeperProperties = new GetZookeeperProperties();
        getZookeeperProperties.getPropertiesByName();
        zookeeperId = getZookeeperProperties.getZookeeperIpPort();
        List<ConsumerMetaData> consumerMetaDataList = new ArrayList<>();
        ZooKeeper zooKeeperConnection =null;
        CountDownLatch countDownLatch = new CountDownLatch(1);
        Watcher watcher =new Watcher() {
            @Override
            public void process(WatchedEvent watchedEvent) {
                if(watchedEvent.getState() == Event.KeeperState.SyncConnected){
                    System.out.println("消费者zookeeper连接创建成功！");
                    countDownLatch.countDown();
                }else if (watchedEvent.getState()== Event.KeeperState.Disconnected){
                    System.out.println("消费者zookeeper连接失败");
                }else if (watchedEvent.getState() == Event.KeeperState.Expired){
                    System.out.println("消费者zookeeper会话超时");
                }
            }
        };
        zooKeeperConnection = new ZooKeeper(zookeeperId, 20000, watcher);
        countDownLatch.await();
        List<String> consumerGroupList = zooKeeperConnection.getChildren(CONSUMERS, false);
        for (String consumerGroup : consumerGroupList) {
            List<String> consumerList = zooKeeperConnection.getChildren(CONSUMERS + "/" + consumerGroup + IDS, false);
            for (String consumerId : consumerList) {
                byte[] data = zooKeeperConnection.getData(CONSUMERS + "/" + consumerGroup + IDS + "/" + consumerId, false, null);
                ConsumerMetaData consumerMetaData =new ConsumerMetaData();

                consumerMetaData =getConsumerMetaData(consumerGroup,new String(data),zooKeeperConnection);
                consumerMetaDataList.add(consumerMetaData);
            }
        }
        zooKeeperConnection.close();
        return consumerMetaDataList;
    }

    public static void main(String[] args) throws Exception {
        ConsumerConnectZookeeper consumerConnectZookeeper = new ConsumerConnectZookeeper();
        ZooKeeper zooKeeperConnection = consumerConnectZookeeper.connect();

        //获取所有的消费者元数据信息
        GetConsumerMetaDataImpl getConsumerMetaData =new GetConsumerMetaDataImpl();
        List<ConsumerMetaData> consumerMetaData = getConsumerMetaData.getConsumerMetaData();
        System.out.println(consumerMetaData);
        // 获取某一个消费者的元数据信息
        ConsumerMetaData consumerMetaData1 = getConsumerMetaData.getConsumerMetaData("consumerGroup001", "consumer001", zooKeeperConnection);
        System.out.println(consumerMetaData1);
    }
}
